package com.shujia.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Code03Function {
  def main(args: Array[String]): Unit = {
    /**
     * Spark中的常用算子：
     * 算子类型：
     * 转换算子（Transformation）
     *    可以再由返回的RDD泛型，可以分为Value类型和KeyValue类型 
     * 特征：可以返回一个RDD,但是不会触发任务
     *
     * 行动算子 （Action ）
     * 特征：不会返回一个RDD，但是可以创建Job
     *
     */

    val conf: SparkConf = new SparkConf()
    conf.setAppName("WordCount")
    conf.setMaster("local") // 设置运行模式为本地模式

    val sc = new SparkContext(conf) // 获取操作类的对象 -> 通常称为 sc

    // 2.获取操作对象后，再读取数据
    //   读取数据后返回的是一个RDD其中 数据格式泛型为String 类比 Scala中的List[String] 注意：RDD不是一个容器，并不会保存数据
    //   minPartitions 最小分区数据表示限定下限 那么对于切片数如果和最小分区数不匹配，那么会以倍数形式增加
    val wordFileData: RDD[String] = sc.textFile("spark_code/data/word", 4)
    println("wordFileDataRDDPartition:" + wordFileData.getNumPartitions)

    // 3.数据计算
    val flatMapWord: RDD[String] = wordFileData
      //    .foreach(x => println("line:" + x))
      .flatMap(
        (line: String) => {
          // java,spark,java,hadoop
          println("flatmap执行了...")
          line.split(",")
        }

      )
    println("flatMapWordRDDPartition:" + flatMapWord.getNumPartitions)


    val firstMap: RDD[String] = flatMapWord.map {
      //  [java,spark,java,hadoop] => 每个单词都会触发一次map计算逻辑
      case word => {
        println("first map running...")
        word
      }
    }


    val groupByWord: RDD[(String, Iterable[String])] = firstMap.groupBy(x => x, 2)

    println("groupByWordRDDPartition:" + groupByWord.getNumPartitions)


    val mapRes: RDD[(String, Int)] = groupByWord.map {
      case (word, wordLine) => {
        println("map执行了...")
        (word, wordLine.size)
      }
    }

    println("mapResRDDPartition:" + mapRes.getNumPartitions)

    while(true){}

  }
}
